package com.ztorn.cdc.doris;

import com.ztorn.cdc.AbstractSinkBuilder;
import com.ztorn.cdc.SinkBuilder;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.FlinkCDCConfig;
import com.ztorn.common.model.Table;
import lombok.extern.slf4j.Slf4j;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.RowDataSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.utils.TypeConversions;

import java.io.Serializable;
import java.util.*;

@Slf4j
public class DorisSinkBuilder extends AbstractSinkBuilder implements Serializable {
    public static final String KEY_WORD = "datastream-doris";
    private static final long serialVersionUID = 8330362249137471854L;

    public DorisSinkBuilder() {}

    public DorisSinkBuilder(FlinkCDCConfig config) {
        super(config);
    }

    @Override
    public String getHandle() {
        return KEY_WORD;
    }

    @Override
    public SinkBuilder create(FlinkCDCConfig config) {
        return new DorisSinkBuilder(config);
    }

//    @Override
//    public void addSink(
//            StreamExecutionEnvironment env,
//            DataStream<RowData> rowDataDataStream,
//            Table table,
//            List<String> columnNameList,
//            List<LogicalType> columnTypeList) {
//
////        DorisExecutionOptions.Builder dorisExecutionOptionsBuilder =
////                DorisExecutionOptions.builder();
////        Map<String, String> sink = config.getSink();
////        if (sink.containsKey("sink.batch.size")) {
////            dorisExecutionOptionsBuilder.setBatchSize(Integer.valueOf(sink.get("sink.batch.size")));
////        }
////        if (sink.containsKey("sink.batch.interval")) {
////            dorisExecutionOptionsBuilder.setBatchIntervalMs(
////                    Long.valueOf(sink.get("sink.batch.interval")));
////        }
////        if (sink.containsKey("sink.max-retries")) {
////            dorisExecutionOptionsBuilder.setMaxRetries(
////                    Integer.valueOf(sink.get("sink.max-retries")));
////        }
////        if (sink.containsKey("sink.enable-delete")) {
////            dorisExecutionOptionsBuilder.setEnableDelete(
////                    Boolean.valueOf(sink.get("sink.enable-delete")));
////        }
////        dorisExecutionOptionsBuilder.setStreamLoadProp(getProperties());
//        DorisSink.Builder<RowData> builder = DorisSink.builder();
//        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
//        dorisBuilder.setFenodes(config.getSink().get("fenodes"))
//                .setTableIdentifier(getSinkSchemaName(table) + "." + getSinkTableName(table))
//                .setUsername(config.getSink().get("username"))
//                .setPassword(config.getSink().get("password"));
//        log.info("========={}",config.getSink().get("fenodes"));
//// json format to streamload
//        Properties properties = new Properties();
//        properties.setProperty("format", "json");
//        properties.setProperty("read_json_by_line", "true");
//        DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
//        executionBuilder.setLabelPrefix("x-doris") //streamload label prefix
//                .setStreamLoadProp(properties); //streamload params
//        final String[] columnNames = columnNameList.toArray(new String[columnNameList.size()]);
//        DataType[] types = {DataTypes.INT(), DataTypes.VARCHAR(50), DataTypes.INT()};
//        builder.setDorisReadOptions(DorisReadOptions.builder().build())
//                .setDorisExecutionOptions(executionBuilder.build())
//                .setSerializer(RowDataSerializer.builder()    //serialize according to rowdata
//                        .setFieldNames(columnNames)
//                        .setType("json")           //json format
//                        .setFieldType(types).build())
//                .setDorisOptions(dorisBuilder.build());
//        DataStream<RowData> source = env.fromElements("")
//                .map(new MapFunction<String, RowData>() {
//                    @Override
//                    public RowData map(String value) throws Exception {
//                        GenericRowData genericRowData = new GenericRowData(4);
//                        genericRowData.setField(0, 1);
//                        genericRowData.setField(1, StringData.fromString("zhangsan"));
//                        genericRowData.setField(2, 25);
//                        return genericRowData;
//                    }
//                });
//        rowDataDataStream.map(new MapFunction<RowData, Object>() {
//            @Override
//            public Object map(RowData rowData) throws Exception {
//                log.info("******************{}",JSON.toJSON(rowData));
//                return null;
//            }
//        });
//        source.sinkTo(builder.build());
//    }

//


    @Override
    public void addSink(
            StreamExecutionEnvironment env,
            DataStream<RowData> rowDataDataStream,
            Table table,
            List<String> columnNameList,
            List<LogicalType> columnTypeList) {

        Map<String, String> sink = config.getSink();

        // Create FieldNames and FieldType for RowDataSerializer.
        final String[] columnNames = columnNameList.toArray(new String[columnNameList.size()]);
        final List<DataType> dataTypeList = new ArrayList<>();
        for (LogicalType logicalType : columnTypeList) {
            dataTypeList.add(TypeConversions.fromLogicalToDataType(logicalType));
        }
        final DataType[] columnTypes = dataTypeList.toArray(new DataType[dataTypeList.size()]);

        // Create DorisReadOptions for DorisSink.
        final DorisReadOptions.Builder readOptionBuilder = DorisReadOptions.builder();
        if (sink.containsKey(DorisSinkOptions.DORIS_DESERIALIZE_ARROW_ASYNC.key())) {
            readOptionBuilder.setDeserializeArrowAsync(
                    Boolean.valueOf(
                            sink.get(DorisSinkOptions.DORIS_DESERIALIZE_ARROW_ASYNC.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_DESERIALIZE_QUEUE_SIZE.key())) {
            readOptionBuilder.setDeserializeQueueSize(
                    Integer.valueOf(sink.get(DorisSinkOptions.DORIS_DESERIALIZE_QUEUE_SIZE.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_EXEC_MEM_LIMIT.key())) {
            readOptionBuilder.setExecMemLimit(
                    Long.valueOf(sink.get(DorisSinkOptions.DORIS_EXEC_MEM_LIMIT.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_FILTER_QUERY.key())) {
            readOptionBuilder.setFilterQuery(
                    String.valueOf(sink.get(DorisSinkOptions.DORIS_FILTER_QUERY.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_READ_FIELD.key())) {
            readOptionBuilder.setReadFields(sink.get(DorisSinkOptions.DORIS_READ_FIELD.key()));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_BATCH_SIZE.key())) {
            readOptionBuilder.setRequestBatchSize(
                    Integer.valueOf(sink.get(DorisSinkOptions.DORIS_BATCH_SIZE.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_REQUEST_CONNECT_TIMEOUT_MS.key())) {
            readOptionBuilder.setRequestConnectTimeoutMs(
                    Integer.valueOf(
                            sink.get(DorisSinkOptions.DORIS_REQUEST_CONNECT_TIMEOUT_MS.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_REQUEST_QUERY_TIMEOUT_S.key())) {
            readOptionBuilder.setRequestQueryTimeoutS(
                    Integer.valueOf(
                            sink.get(DorisSinkOptions.DORIS_REQUEST_QUERY_TIMEOUT_S.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_REQUEST_READ_TIMEOUT_MS.key())) {
            readOptionBuilder.setRequestReadTimeoutMs(
                    Integer.valueOf(
                            sink.get(DorisSinkOptions.DORIS_REQUEST_READ_TIMEOUT_MS.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_REQUEST_RETRIES.key())) {
            readOptionBuilder.setRequestRetries(
                    Integer.valueOf(sink.get(DorisSinkOptions.DORIS_REQUEST_RETRIES.key())));
        }
        if (sink.containsKey(DorisSinkOptions.DORIS_REQUEST_TABLET_SIZE.key())) {
            readOptionBuilder.setRequestTabletSize(
                    Integer.valueOf(sink.get(DorisSinkOptions.DORIS_REQUEST_TABLET_SIZE.key())));
        }

        // Create DorisOptions for DorisSink.
        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
        dorisBuilder
                .setFenodes(config.getSink().get(DorisSinkOptions.FENODES.key()))
                .setTableIdentifier(getSinkSchemaName(table) + "." + getSinkTableName(table))
                .setUsername(config.getSink().get(DorisSinkOptions.USERNAME.key()))
                .setPassword(config.getSink().get(DorisSinkOptions.PASSWORD.key()))
                .build();

        // Create DorisExecutionOptions for DorisSink.
        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
        if (sink.containsKey(DorisSinkOptions.SINK_BUFFER_COUNT.key())) {
            executionBuilder.setBufferCount(
                    Integer.valueOf(sink.get(DorisSinkOptions.SINK_BUFFER_COUNT.key())));
        }
        if (sink.containsKey(DorisSinkOptions.SINK_BUFFER_SIZE.key())) {
            executionBuilder.setBufferSize(
                    Integer.valueOf(sink.get(DorisSinkOptions.SINK_BUFFER_SIZE.key())));
        }
        if (sink.containsKey(DorisSinkOptions.SINK_ENABLE_DELETE.key())) {
            executionBuilder.setDeletable(
                    Boolean.valueOf(sink.get(DorisSinkOptions.SINK_ENABLE_DELETE.key())));
        }
        if (sink.containsKey(DorisSinkOptions.SINK_LABEL_PREFIX.key())) {
            executionBuilder.setLabelPrefix(
                    sink.get(DorisSinkOptions.SINK_LABEL_PREFIX.key())
                            + "-"
                            + getSinkSchemaName(table)
                            + "_"
                            + getSinkTableName(table));
        } else {
            executionBuilder.setLabelPrefix(
                    "ztorn-"
                            + getSinkSchemaName(table)
                            + "_"
                            + getSinkTableName(table)
                            + UUID.randomUUID());
        }
        if (sink.containsKey(DorisSinkOptions.SINK_MAX_RETRIES.key())) {
            executionBuilder.setMaxRetries(
                    Integer.valueOf(sink.get(DorisSinkOptions.SINK_MAX_RETRIES.key())));
        }
        executionBuilder.setStreamLoadProp(getProperties());

        // Create DorisSink.
        DorisSink.Builder<RowData> builder = DorisSink.builder();
        builder.setDorisReadOptions(readOptionBuilder.build())
                .setDorisExecutionOptions(executionBuilder.build())
                .setSerializer(
                        RowDataSerializer.builder()
                                .setFieldNames(columnNames)
                                .setType("json")
                                .setFieldType(columnTypes)
                                .build())
                .setDorisOptions(dorisBuilder.build());

        rowDataDataStream
                .sinkTo(builder.build())
                .name(
                        "Doris Sink(table=["
                                + getSinkSchemaName(table)
                                + "."
                                + getSinkTableName(table)
                                + "])");
    }

    @Override
    protected Properties getProperties() {
        Properties properties = new Properties();
        Map<String, String> sink = config.getSink();
        for (Map.Entry<String, String> entry : sink.entrySet()) {
            if (Asserts.isNotNullString(entry.getKey())
                    && entry.getKey().startsWith("sink.properties")
                    && Asserts.isNotNullString(entry.getValue())) {
                properties.setProperty(
                        entry.getKey().replace("sink.properties.", ""), entry.getValue());
            }
        }
        return properties;
    }
}
